package cn.llyong.rpc.server.handler;

import cn.llyong.rpc.common.bo.RpcRequest;
import cn.llyong.rpc.common.bo.RpcResponse;
import cn.llyong.rpc.common.enumtype.RpcResponseCode;
import cn.llyong.rpc.common.exception.RpcException;
import cn.llyong.rpc.common.utils.RpcCacheHolder;
import com.alibaba.fastjson.JSON;
import com.baidu.bjf.remoting.protobuf.Codec;
import com.baidu.bjf.remoting.protobuf.ProtobufProxy;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * @description: 
 * @author: lvyong
 * @date: 2020年06月30日 00:47
 */
public class RpcServerHandler extends ChannelDuplexHandler {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    public static ConcurrentMap<String,ChannelHandlerContext> nettyClients = new ConcurrentHashMap<String,ChannelHandlerContext>();

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        RpcResponse response = new RpcResponse();
        try {
            Codec<RpcRequest> requestCodec = ProtobufProxy.create(RpcRequest.class);

            ByteBuf buf = (ByteBuf)msg;
            byte[] req = new byte[buf.readableBytes()];
            buf.readBytes(req);
            RpcRequest request = requestCodec.decode(req);

            logger.info("服务端收到请求对象: {}", request);

            response.setRequestId(request.getRequestId());
            response.setCode(RpcResponseCode.CODE_200.getCode());
            response.setMsg(RpcResponseCode.CODE_200.getMsg());
            String className = request.getServiceItf();
            Object target = RpcCacheHolder.PROVIDERS_SERVICES.get(className);
            if (target == null) {
                throw new RpcException("没有可用的服务！！！");
            }

            List<RpcRequest.Param> params = request.getData();
            int length = params == null ? 0 : params.size();
            Class<?>[] parameterTypes = new Class[length];
            Object[] parameters = new Object[length];
            if (params != null && params.size() > 0) {
                for (int i = 0; i < params.size(); i++) {
                    RpcRequest.Param param = params.get(i);
                    parameterTypes[i] = Class.forName(param.getType());
                    if (param.getData() != null && param.getData() != "") {
                        RpcRequest.Param p = JSON.parseObject(param.getData(), RpcRequest.Param.class);
                        Class<?> clazz = Class.forName(p.getType());
                        parameters[i] = JSON.parseObject(p.getData(), clazz);
                    }
                }
            }

            Method method = target.getClass().getMethod(request.getMethodName(), parameterTypes);
            Object rt = method.invoke(target, parameters);
            if (rt != null) {
                String jsonString = JSON.toJSONString(rt);
                RpcResponse.Result result = new RpcResponse.Result();
                result.setType(method.getReturnType().getName());
                result.setData(jsonString);
                response.setData(result);
            }
        }catch (Exception e){
            logger.error("Rpc execute Error!", e);
            response.setCode(RpcResponseCode.CODE_500.getCode());
            response.setMsg(e.getMessage());
        }
        try {
            Codec<RpcResponse> responseCodec = ProtobufProxy.create(RpcResponse.class);
            byte[] encode = responseCodec.encode(response);
            ctx.writeAndFlush(Unpooled.copiedBuffer(encode));
        } catch (Exception e) {
            logger.error("返回对象序列化失败！！！", e);
        }
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        String channelId=ctx.channel().id().toString();
        logger.info("client connect;address: {}, channelId: {}", ctx.channel().remoteAddress(), channelId);
        nettyClients.put(channelId, ctx);
        logger.info("client count: {}", nettyClients.size());
    }
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {

    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)  {
        ctx.close();
        String channelId = ctx.channel().id().toString();
        nettyClients.remove(channelId);
        logger.error("catch exception！！！remove channel id:【{}】,client count: {}", channelId, nettyClients.size());
        logger.error("系统异常！！！", cause);
    }
    @Override
    public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise){
        ctx.disconnect(promise);
        String channelId = ctx.channel().id().toString();
        nettyClients.remove(channelId);
        logger.info("disconnect！！！remove channel id:【{}】,client count: {}", channelId, nettyClients.size());
    }

    @Override
    public void close(ChannelHandlerContext ctx, ChannelPromise promise)  {
        ctx.close(promise);
        String channelId = ctx.channel().id().toString();
        nettyClients.remove(channelId);
        logger.info("close channel！！！remove channel id:【{}】,client count: {}", channelId, nettyClients.size());
    }
}
